package com.bootlever.framework.rocketmq.templates;

import com.alibaba.rocketmq.client.ClientConfig;
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.MQConsumer;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MQProducer;
import com.alibaba.rocketmq.client.producer.TransactionMQProducer;
import com.bootlever.framework.common.utils.StringUtils;
import com.bootlever.framework.rocketmq.properties.RocketmqProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class RocketmqTemplate {

    private static final Logger logger = LoggerFactory.getLogger(RocketmqTemplate.class);

    Map<String, MQProducer> producers = new ConcurrentHashMap<String, MQProducer>();
    Map<String, MQConsumer> consumers = new ConcurrentHashMap<String, MQConsumer>();

    private RocketmqProperties rocketmqProperties;

    public RocketmqTemplate(RocketmqProperties rocketmqProperties) {
        this.rocketmqProperties = rocketmqProperties;
    }

    public MQProducer getDefaultMQProducer(String producerGroup) throws Exception {
        if (StringUtils.isBlank(producerGroup)) {
            logger.error("producerGroup为空");
            throw new Exception("producerGroup is empty");
        }
        if (producers.get(producerGroup) != null){
            return producers.get(producerGroup);
        }
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(producerGroup);
        defaultMQProducer = (DefaultMQProducer) initNameSrv(defaultMQProducer);
        producers.putIfAbsent(producerGroup, defaultMQProducer);
        return defaultMQProducer;
    }

    public MQProducer getTransactionMQProducer(String producerGroup) throws Exception {
        if (StringUtils.isBlank(producerGroup)) {
            logger.error("producerGroup为空");
            throw new Exception("producerGroup is empty");
        }
        if (producers.get(producerGroup) != null){
            return producers.get(producerGroup);
        }
        TransactionMQProducer producer = new TransactionMQProducer(producerGroup);
        producer = (TransactionMQProducer) initNameSrv(producer);
        producers.putIfAbsent(producerGroup, producer);
        return producer;
    }

    public MQConsumer getDefaultMQPushConsumer(String consumerGroup) throws Exception {
        if (StringUtils.isBlank(consumerGroup)) {
            logger.error("consumerGroup为空");
            throw new Exception("consumerGroup is empty");
        }
        if (consumers.get(consumerGroup) != null) {
            return consumers.get(consumerGroup);
        }
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
        defaultMQPushConsumer = (DefaultMQPushConsumer) initNameSrv(defaultMQPushConsumer);
        consumers.putIfAbsent(consumerGroup, defaultMQPushConsumer);
        return defaultMQPushConsumer;
    }

    public MQConsumer getDefaultMQPullConsumer(String consumerGroup) throws Exception {
        if (StringUtils.isBlank(consumerGroup)) {
            logger.error("consumerGroup为空");
            throw new Exception("consumerGroup is empty");
        }
        if (consumers.get(consumerGroup) != null) {
            return consumers.get(consumerGroup);
        }
        DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer(consumerGroup);
        defaultMQPullConsumer = (DefaultMQPullConsumer) initNameSrv(defaultMQPullConsumer);
        consumers.putIfAbsent(consumerGroup, defaultMQPullConsumer);
        return defaultMQPullConsumer;
    }


    private ClientConfig initNameSrv(ClientConfig clientConfig){
        if (StringUtils.isNotBlank(rocketmqProperties.getNamesrvAddr())){
            clientConfig.setNamesrvAddr(rocketmqProperties.getNamesrvAddr());
        }
        if (StringUtils.isNotBlank(rocketmqProperties.getInstanceName())){
            clientConfig.setInstanceName(rocketmqProperties.getInstanceName());
        }
        if (null != rocketmqProperties.getClientCallbackExecutorThreads()){
            clientConfig.setClientCallbackExecutorThreads(rocketmqProperties.getClientCallbackExecutorThreads());
        }
        if (null != rocketmqProperties.getPollNameServerInteval()){
            clientConfig.setPollNameServerInteval(rocketmqProperties.getPollNameServerInteval());
        }
        if (null != rocketmqProperties.getHeartbeatBrokerInterval()) {
            clientConfig.setHeartbeatBrokerInterval(rocketmqProperties.getHeartbeatBrokerInterval());
        }
        if (null != rocketmqProperties.getPersistConsumerOffsetInterval()) {
            clientConfig.setPersistConsumerOffsetInterval(rocketmqProperties.getPersistConsumerOffsetInterval());
        }
        return clientConfig;
    }
}
